package cn.lecosa.spark.hbase
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Connection
import org.apache.hadoop.hbase.HConstants
import org.apache.hadoop.hbase.client.ConnectionFactory
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.SQLContext

object ReadHbase {
  def main(args: Array[String]) {
    val conf = new SparkConf(); //创建SparkConf对象  设置spark参数
    conf.setAppName("wow,My First Spark App!"); //设置应用程序的名称，在spark程序运行的监控界面可以看到名称
    conf.setMaster("local[2]"); //此时，程序在本地运行，不需要安装spark集群
    val sc = new SparkContext(conf);

    val hiveContext = new HiveContext(sc);

    val hbaseconf = HBaseConfiguration.create(sc.hadoopConfiguration)

    //设置查询的表名
    hbaseconf.set(TableInputFormat.INPUT_TABLE, "emp1")

    val usersRDD = sc.newAPIHadoopRDD(hbaseconf, classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])

    usersRDD.cache()
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    val count = usersRDD.count()
    println("Users RDD Count:" + count)

    //    对象 存 队列取出来  校验hbase 

    //    hive 外部表 

    //    hiveContext.sql("show databases").show()
    //    hiveContext.sql("use lecosa").show()
    //    hiveContext.sql("show tables").show()

    val shop = usersRDD.map(r => (
      Bytes.toString(r._2.getRow),
      Bytes.toString(r._2.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("c1"))),
      Bytes.toString(r._2.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("c2"))),
      Bytes.toString(r._2.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("c3"))),
      Bytes.toString(r._2.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("c4"))),
      Bytes.toString(r._2.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("c5"))),
      Bytes.toString(r._2.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("c6")))
    )).toDF()

    shop.registerTempTable("shop")

    // 测试
//    val df2 = sqlContext.sql("create table lecosa.tab_x_3  as SELECT * FROM shop")
       val df2 = sqlContext.sql(" SELECT * FROM shop").show()
//    df2.printSchema()

    //遍历输出
    /*    val rdd1 = usersRDD.foreach {
      case (_, result) => {
        //获取行键
        val key = Bytes.toString(result.getRow)
        //通过列族和列名获取列
        val c1 = Bytes.toString(result.getValue("cf1".getBytes, "c1".getBytes))
        val c2 = Bytes.toString(result.getValue("cf1".getBytes, "c2".getBytes))
        //      println("key======>>>"+key +c1);
        println("Row key:" + key + " c1:" + c1 + " c2:" + c2)
        val sql = s"insert into lecosa.tab_x_1  values('${key}','${c1}','${c2}')  "
        println(sql)
        hiveContext.sql(sql)
      }
    }*/

    sc.stop()
  }
}


